/*-------------------------------------------------------------------------
 *
 * logicalfuncs.c
 *
 *	   用于使用逻辑解码和管理逻辑复制插槽的支持函数。
 *
 * Copyright (c) 2012-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/logical/logicalfuncs.c
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include <unistd.h>

#include "access/xact.h"
#include "access/xlog_internal.h"
#include "access/xlogrecovery.h"
#include "access/xlogutils.h"
#include "catalog/pg_type.h"
#include "fmgr.h"
#include "funcapi.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/message.h"
#include "storage/fd.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/regproc.h"
#include "utils/resowner.h"

/* 写出数据的私有数据 */
typedef struct DecodingOutputState
{
	Tuplestorestate *tupstore;
	TupleDesc	tupdesc;
	bool		binary_output;
	int64		returned_rows;
} DecodingOutputState;

/*
 * 准备输出插件的写入。
 */
static void fc_LogicalOutputPrepareWrite(LogicalDecodingContext *fc_ctx, XLogRecPtr fc_lsn, TransactionId fc_xid,
						  bool fc_last_write)
{
	resetStringInfo(fc_ctx->out);
}

/*
 * 将输出插件写入tuplestore。
 */
static void fc_LogicalOutputWrite(LogicalDecodingContext *fc_ctx, XLogRecPtr fc_lsn, TransactionId fc_xid,
				   bool fc_last_write)
{
	Datum		fc_values[3];
	bool		fc_nulls[3];
	DecodingOutputState *fc_p;

	/* SQL Datums的长度是有限的... */
	if (fc_ctx->out->len > MaxAllocSize - VARHDRSZ)
		elog(ERROR, "too much output for sql interface");

	fc_p = (DecodingOutputState *) fc_ctx->output_writer_private;

	memset(fc_nulls, 0, sizeof(fc_nulls));
	fc_values[0] = LSNGetDatum(fc_lsn);
	fc_values[1] = TransactionIdGetDatum(fc_xid);

	/*
	 * 在我们写入文本输出时，确保ctx->out是数据库编码。
	 */
	if (!fc_p->binary_output)
		Assert(pg_verify_mbstr(GetDatabaseEncoding(),
							   fc_ctx->out->data, fc_ctx->out->len,
							   false));

	/* 真讨厌，但cstring_to_text_with_len对bytea完全有效 */
	fc_values[2] = PointerGetDatum(cstring_to_text_with_len(fc_ctx->out->data, fc_ctx->out->len));

	tuplestore_putvalues(fc_p->tupstore, fc_p->tupdesc, fc_values, fc_nulls);
	fc_p->returned_rows++;
}

/*
 * 用于各种SQL可调用逻辑解码函数的辅助函数。
 */
static Datum fc_pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool fc_confirm, bool fc_binary)
{
	Name		fc_name;
	XLogRecPtr	fc_upto_lsn;
	int32		fc_upto_nchanges;
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
	MemoryContext fc_per_query_ctx;
	MemoryContext fc_oldcontext;
	XLogRecPtr	fc_end_of_wal;
	LogicalDecodingContext *fc_ctx;
	ResourceOwner fc_old_resowner = CurrentResourceOwner;
	ArrayType  *fc_arr;
	Size		fc_ndim;
	List	   *fc_options = NIL;
	DecodingOutputState *fc_p;

	CheckSlotPermissions();

	CheckLogicalDecodingRequirements();

	if (PG_ARGISNULL(0))
		ereport(ERROR,
				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
				 errmsg("slot name must not be null")));
	fc_name = PG_GETARG_NAME(0);

	if (PG_ARGISNULL(1))
		fc_upto_lsn = InvalidXLogRecPtr;
	else
		fc_upto_lsn = PG_GETARG_LSN(1);

	if (PG_ARGISNULL(2))
		fc_upto_nchanges = InvalidXLogRecPtr;
	else
		fc_upto_nchanges = PG_GETARG_INT32(2);

	if (PG_ARGISNULL(3))
		ereport(ERROR,
				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
				 errmsg("options array must not be null")));
	fc_arr = PG_GETARG_ARRAYTYPE_P(3);

	/* 要写入输出的状态 */
	fc_p = palloc0(sizeof(DecodingOutputState));

	fc_p->binary_output = fc_binary;

	fc_per_query_ctx = fc_rsinfo->econtext->ecxt_per_query_memory;
	fc_oldcontext = MemoryContextSwitchTo(fc_per_query_ctx);

	/* 解构选项数组 */
	fc_ndim = ARR_NDIM(fc_arr);
	if (fc_ndim > 1)
	{
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("array must be one-dimensional")));
	}
	else if (array_contains_nulls(fc_arr))
	{
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("array must not contain nulls")));
	}
	else if (fc_ndim == 1)
	{
		int			fc_nelems;
		Datum	   *fc_datum_opts;
		int			fc_i;

		Assert(ARR_ELEMTYPE(fc_arr) == TEXTOID);

		deconstruct_array(fc_arr, TEXTOID, -1, false, TYPALIGN_INT,
						  &fc_datum_opts, NULL, &fc_nelems);

		if (fc_nelems % 2 != 0)
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("array must have even number of elements")));

		for (fc_i = 0; fc_i < fc_nelems; fc_i += 2)
		{
			char	   *fc_name = TextDatumGetCString(fc_datum_opts[fc_i]);
			char	   *fc_opt = TextDatumGetCString(fc_datum_opts[fc_i + 1]);

			fc_options = lappend(fc_options, makeDefElem(fc_name, (Node *) makeString(fc_opt), -1));
		}
	}

	InitMaterializedSRF(fcinfo, 0);
	fc_p->tupstore = fc_rsinfo->setResult;
	fc_p->tupdesc = fc_rsinfo->setDesc;

	/*
	 * 计算当前的wal结束点。
	 */
	if (!RecoveryInProgress())
		fc_end_of_wal = GetFlushRecPtr(NULL);
	else
		fc_end_of_wal = GetXLogReplayRecPtr(NULL);

	ReplicationSlotAcquire(NameStr(*fc_name), true);

	PG_TRY();
	{
		/* 在槽的confirmed_flush处重新启动 */
		fc_ctx = CreateDecodingContext(InvalidXLogRecPtr,
									fc_options,
									false,
									XL_ROUTINE(.page_read = read_local_xlog_page,
											   .segment_open = wal_segment_open,
											   .segment_close = wal_segment_close),
									fc_LogicalOutputPrepareWrite,
									fc_LogicalOutputWrite, NULL);

		/*
		 * 在CreateDecodingContext中的完整性检查后，确保restart_lsn是有效的。
		 * 避免在此错误信息中使用“无法获取更改”的措辞，因为这会让人混淆不清楚没有可用的更改。
		 */
		if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("can no longer get changes from replication slot \"%s\"",
							NameStr(*fc_name)),
					 errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));

		MemoryContextSwitchTo(fc_oldcontext);

		/*
		 * 检查输出插件是否写入文本输出，如果这是我们所需要的。
		 */
		if (!fc_binary &&
			fc_ctx->options.output_type !=OUTPUT_PLUGIN_TEXTUAL_OUTPUT)
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("logical decoding output plugin \"%s\" produces binary output, but function \"%s\" expects textual data",
							NameStr(MyReplicationSlot->data.plugin),
							format_procedure(fcinfo->flinfo->fn_oid))));

		fc_ctx->output_writer_private = fc_p;

		/*
		 * WAL的解码必须从restart_lsn开始，以便能够将所有在槽的confirmed_flush后提交的xacts
		 * 累积到重排序缓冲区中。
		 */
		XLogBeginRead(fc_ctx->reader, MyReplicationSlot->data.restart_lsn);

		/* 无效的非时间旅行条目 */
		InvalidateSystemCaches();

		/* 解码直到记录用完 */
		while (fc_ctx->reader->EndRecPtr < fc_end_of_wal)
		{
			XLogRecord *fc_record;
			char	   *fc_errm = NULL;

			fc_record = XLogReadRecord(fc_ctx->reader, &fc_errm);
			if (fc_errm)
				elog(ERROR, "could not find record for logical decoding: %s", fc_errm);

			/*
			 * 以上的{begin_txn,change,commit_txn}_wrapper回调将
			 * 描述存储到我们的tuplestore中。
			 */
			if (fc_record != NULL)
				LogicalDecodingProcessRecord(fc_ctx, fc_ctx->reader);

			/* 检查限制 */
			if (fc_upto_lsn != InvalidXLogRecPtr &&
				fc_upto_lsn <= fc_ctx->reader->EndRecPtr)
				break;
			if (fc_upto_nchanges != 0 &&
				fc_upto_nchanges <= fc_p->returned_rows)
				break;
			CHECK_FOR_INTERRUPTS();
		}

		/*
		 * 逻辑解码可能在事务管理期间覆盖了 CurrentResourceOwner，
		 * 因此需要恢复执行者的值。（这是一种应急措施，但现在不值得清理。）
		 */
		CurrentResourceOwner = fc_old_resowner;

		/*
		 * 下次，从我们离开的地方开始。（狩猎事情，家庭生意..）
		 */
		if (fc_ctx->reader->EndRecPtr != InvalidXLogRecPtr && fc_confirm)
		{
			LogicalConfirmReceivedLocation(fc_ctx->reader->EndRecPtr);

			/*
			 * 如果只有confirmed_flush_lsn改变，槽不会被上面的标记为脏。
			 * walsender接口的调用者预计会跟踪自己的进度， 
			 * 并且不需要写出来。但SQL接口用户不能
			 * 指定自己的起始位置，跟踪进度会更加困难，因此我们应该
			 * 更加努力地为他们保存它。
			 *
			 * 将槽标记为脏，这样在下一个检查点时会被写出。
			 * 在崩溃时我们仍然会失去它的位置，如文档所述， 
			 * 但这总比在干净重启时总是失去位置要好。
			 */
			ReplicationSlotMarkDirty();
		}

		/* 释放上下文，调用关闭回调 */
		FreeDecodingContext(fc_ctx);

		ReplicationSlotRelease();
		InvalidateSystemCaches();
	}
	PG_CATCH();
	{
		/* 清除所有时间旅行条目 */
		InvalidateSystemCaches();

		PG_RE_THROW();
	}
	PG_END_TRY();

	return (Datum) 0;
}

/*
 * SQL函数将更改流作为文本返回，消耗数据。
 */
Datum pg_logical_slot_get_changes(PG_FUNCTION_ARGS)
{
	return fc_pg_logical_slot_get_changes_guts(fcinfo, true, false);
}

/*
 * SQL函数将更改流作为文本返回，仅向前查看。
 */
Datum pg_logical_slot_peek_changes(PG_FUNCTION_ARGS)
{
	return fc_pg_logical_slot_get_changes_guts(fcinfo, false, false);
}

/*
 * SQL函数将更改流以二进制形式返回，消耗数据。
 */
Datum pg_logical_slot_get_binary_changes(PG_FUNCTION_ARGS)
{
	return fc_pg_logical_slot_get_changes_guts(fcinfo, true, true);
}

/*
 * SQL函数将更改流以二进制形式返回，仅向前查看。
 */
Datum pg_logical_slot_peek_binary_changes(PG_FUNCTION_ARGS)
{
	return fc_pg_logical_slot_get_changes_guts(fcinfo, false, true);
}


/*
 * SQL函数用于将逻辑解码消息写入WAL。
 */
Datum pg_logical_emit_message_bytea(PG_FUNCTION_ARGS)
{
	bool		fc_transactional = PG_GETARG_BOOL(0);
	char	   *fc_prefix = text_to_cstring(PG_GETARG_TEXT_PP(1));
	bytea	   *fc_data = PG_GETARG_BYTEA_PP(2);
	XLogRecPtr	fc_lsn;

	fc_lsn = LogLogicalMessage(fc_prefix, VARDATA_ANY(fc_data), VARSIZE_ANY_EXHDR(fc_data),
							fc_transactional);
	PG_RETURN_LSN(fc_lsn);
}

Datum pg_logical_emit_message_text(PG_FUNCTION_ARGS)
{
	/* bytea和text是兼容的 */
	return pg_logical_emit_message_bytea(fcinfo);
}
